/* Copyright  2009
   @Author
   Richard Changde Yin            e-mail yinchangde@hotmail.com

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

/**
 * @filename: chunk.cc
 */

#include<server_includes.h>
#include<allocator.h>
#include<database.h>

#include<os.h>
#include<debug.h>
#include<read-config.h>
#include<catalogtables.h>

#include <gettext.h>
#include <error.h>
#include <errmsg_print.h>
#include <util.h>

#include <session.h>
#include <current_session.h>

using namespace BackEnd;

// sets the size of the Chunk allocator for fixed size
// allocator
// we need one integer to store book keeping information
// whether the storage allocation unit is used of not
// when it is deleted this flag is only set to unused
void Chunk::setSize(size_t size)
{
    size_t needSize = size + sizeof(int);
    size_t multiple = (size_t) OS::floor(needSize/sizeof(size_t));
    size_t rem = needSize % sizeof(size_t);

    if (!rem)
        allocSize_ = needSize;
    else
        allocSize_  = (multiple + 1) * sizeof(size_t);
}

void* Chunk::allocateForLargeDataSize (Database *db)
{
    PageInfo* pageInfo    = ((PageInfo*)curPage_);
    BackEnd::DbRetVal ret = db->getAllocDatabaseMutex();
    if (ret != 0)
       return NULL;

    //check whether we have space in curPage
    if (pageInfo->hasFreeSpace_ == 1)
    {
       /// OK, We found this Page has free space.

        unsigned char *data = ((unsigned char*)curPage_) + sizeof(PageInfo);
        pageInfo->hasFreeSpace_ =0;
        *((int*)data) = 1;
        db->releaseAllocDatabaseMutex();
        return data + sizeof(int);
    }
    //no space in current page , get new page from database
    pageInfo = (PageInfo*)db->getFreePage(allocSize_);
      if (!pageInfo)
    {
        db->releaseAllocDatabaseMutex();
        return NULL;
    }
    size_t multiple = (size_t)OS::floor(allocSize_ / PAGE_SIZE);
    size_t offset = ((multiple + 1) * PAGE_SIZE);

    pageInfo->setPageAsUsed(offset);

    //create the link
   ((PageInfo*)curPage_)->nextPage_ = (Page*) pageInfo;
    //Make this as current page
    curPage_   = (Page*) pageInfo;
    unsigned char* data = ((unsigned char*)curPage_) + sizeof(PageInfo);
    //TODO::check whether it is locked
    *((int*)data) = 1;

    db->releaseAllocDatabaseMutex();
    return data + sizeof(int);

}

void* Chunk::allocateFromFirstPage(Database*, int noOfDataNodes)
{
    PageInfo *pageIter = ((PageInfo*)firstPage_);
    unsigned char *data = NULL;
    int i = 0;
    //scan from first page to locate a free node available
    while(NULL != pageIter)
    {
        data = ((unsigned char*)pageIter) + sizeof(PageInfo);
        if (pageIter->hasFreeSpace_ == 1)
        {
            for (i = 0; i< noOfDataNodes -1; i++)
          {
             if (1 == *((int*)data))
               data = data + allocSize_;
             else break;
          }
          if (i != noOfDataNodes -1) break;
        }
        pageIter = (PageInfo*)((PageInfo*) pageIter)->nextPage_;
    }
    if (!pageIter) return NULL;
    *((int*)data) = 1;
    return data + sizeof(int);

}


void* Chunk::allocateFromNewPage(Database *db)
{
    DbRetVal ret = db->getAllocDatabaseMutex();
    if (ret != 0)
       return NULL;

    //get a new page from db
    Page *page = db->getFreePage();
    if (page == NULL)
    {
        db->releaseAllocDatabaseMutex();
        return NULL;
    }
    //Initialize pageInfo for this new page
    PageInfo *pInfo = (PageInfo*)page;
    /// Mark this page is not used.
    pInfo->setPageAsUsed(0);

    //create the link between old page and the newly created page
    PageInfo* pageInfo = ((PageInfo*)curPage_);
    pageInfo->nextPage_ = page;

    //make this new page as the current page
    curPage_ = page;

    unsigned char* data = ((unsigned char*)page) + sizeof(PageInfo);
    *((int*)data) = 1;
    db->releaseAllocDatabaseMutex();
    return data + sizeof(int);
}

//Allocates memory to store data
//TODO::check whether it is locked before allocating.
//delete tuple will set the usedflag to true, but locks will be held
//till commit and it shall be rollback.So make sure that it does not
//allocate deleted tuple which is yet to be committed.

void* Chunk::allocate(Database *db, DbRetVal *status)
{
    PageInfo* pageInfo = ((PageInfo*)curPage_);

    int32_t noOfDataNodes=0;
    if(allocSize_)
      noOfDataNodes=(int32_t)OS::floor((PAGE_SIZE - sizeof(PageInfo))/allocSize_);
    unsigned char *data = ((unsigned char*)curPage_) + sizeof(PageInfo);


    //1.scan through data list and find if any is free to use in current page
    //2.If there is none then
    //    a) get new free page from db. set the prev->next to point
    //       to this new page
    //4.  b) initialize the free page to zero and get first data ptr.
    //5.If there is one, return that

    //For allocation which is greater than PAGE_SIZE
    if (!noOfDataNodes)
    {
      data = (unsigned char*) allocateForLargeDataSize(db);
      return data;
    }
    int _procsot = (current_session)?current_session->getProcSlot():0;
    int32_t ret = getChunkMutex(_procsot);
    if (ret != 0)
    {
        if (status != NULL) *status = (DbRetVal)(ret);
        return NULL;
    }
    int32_t i = noOfDataNodes;
    if (pageInfo->hasFreeSpace_ == 1)
    {
        for (i = 1; i< noOfDataNodes; i++)
        {
          if (*((int*)data) == 1)  data = data + allocSize_;
            else break;
        }
    }
    //It comes here if the pageInfo->hasFreeSpace ==0
    //or there are no free data space in this page
    if (i == noOfDataNodes && *((int*)data) == 1)
    {
        pageInfo->hasFreeSpace_ = 0;
        if (chunkID_ == LockTableId || chunkID_ == TransHasTableId)
        {
           data = (unsigned char*) allocateFromFirstPage(db, noOfDataNodes);
           if (NULL == data)
           {
              data = (unsigned char*) allocateFromNewPage(db);
              if (data == NULL)
              {
                if (status != NULL) *status =(DbRetVal) HA_ERR_OUT_OF_MEM;
              }
           }
        }else
        {
           data = (unsigned char*) allocateFromNewPage(db);
           if (NULL == data)
           {
              data = (unsigned char*) allocateFromFirstPage(db, noOfDataNodes);
              if (data == NULL)
            {
              if (status != NULL) *status = (DbRetVal)HA_ERR_OUT_OF_MEM;
            }
           }
        }
        releaseChunkMutex(_procsot);
        return data;
    }
    *((int*)data) = 1;
    releaseChunkMutex(_procsot);
    return data + sizeof(int);
}


void* Chunk::allocateForLargeDataSize(Database *db, size_t size)
{
    //no need to take chunk mutexes for this, as we are taking alloc database mutex
    size_t multiple = OS::floor(size / PAGE_SIZE);
    size_t offset = ((multiple + 1) * PAGE_SIZE);
    PageInfo* pageInfo = ((PageInfo*)curPage_);
    DbRetVal ret = db->getAllocDatabaseMutex();
    if (ret != 0)
    {
      return NULL;
    }
    pageInfo = (PageInfo*)db->getFreePage(allocSize_);

     if(0==allocSize_)
         pageInfo = (PageInfo*)db->getFreePage(size);
     else
         pageInfo = (PageInfo*)db->getFreePage(allocSize_);


    if (!pageInfo)
    {
        db->releaseAllocDatabaseMutex();
        return NULL;
    }
    //TODO:: logic pending

     if(allocSize_!=0){
           //large size allocate for FixedSize data
          pageInfo->nextPageAfterMerge_ = ((char*)pageInfo + offset);
          ((PageInfo*)curPage_)->nextPage_ = (Page*) pageInfo;
          curPage_ = (Page*) pageInfo;
          unsigned char* data = ((unsigned char*)curPage_) + sizeof(PageInfo);
          *((int*)data) = 1;
          pageInfo->isUsed_=1;
          pageInfo->hasFreeSpace_ = 0;
          db->releaseAllocDatabaseMutex();
          return data + sizeof(int);
     }else{
            //large size allocate for varSize data
            VarSizeInfo *varInfo =
                          (VarSizeInfo*)(((char*)pageInfo) + sizeof(PageInfo));
            pageInfo->nextPageAfterMerge_ = ((char*)pageInfo + offset);
            ((PageInfo*)curPage_)->nextPage_ = (Page*) pageInfo;
            curPage_ = (Page*) pageInfo;
            varInfo->size_= size;
            varInfo->isUsed_= 1;
            pageInfo->isUsed_=1;
            db->releaseAllocDatabaseMutex();
            return (unsigned char *) varInfo + sizeof(VarSizeInfo);
    }
    //REDESIGN MAY BE REQUIRED:Lets us live with this for now.
    //what happens to the space lets say 10000 bytes is allocated
    //it needs 2 pages,= 16000 bytes, 6000 bytes should not be wasted
    //in this case.So need to take of this.
    //Will be coded at later stage as this is developed to support
    //undo logging and currently we shall assume that the logs generated
    //wont be greater than PAGE_SIZE.
    db->releaseAllocDatabaseMutex();
    return NULL;

}

///  Pick up one freed Page for Var Size.

void* Chunk::allocFromNewPageForVarSize(Database *db, size_t size)
{
    //Should be called only for data items <PAGE_SIZE
    DbRetVal ret = db->getAllocDatabaseMutex();
    if (ret != 0)
       return NULL;


    void *vnode = varSizeFirstFitAllocate(size);
    if (vnode != NULL)
    {
       db->releaseAllocDatabaseMutex();
       return vnode;
    }

    Page *newPage = db->getFreePage();

    db->releaseAllocDatabaseMutex();

    if (!newPage)
    {
      return NULL;
    }
    PageInfo *pInfo = (PageInfo*) newPage;
    pInfo->setPageAsUsed(0);
    createDataBucket(newPage, PAGE_SIZE, size);

    ((PageInfo*)curPage_)->nextPage_ = newPage;
    curPage_ = newPage;
    ////////////////////////////////////////////////////////////////////////
    //   |<-- newPage                           |<-- data
    //   ------------------------------------------------------------------
    //   |                 |                    |                       --->
    //   |  PageInfo       |     VarSizeInfo    |                       --->
    //   |-----------------------------------------------------------------
    ////////////////////////////////////////////////////////////////////////
    unsigned char *data= ((unsigned char*)newPage) + sizeof(PageInfo) + sizeof(VarSizeInfo);
    return data;
}

//Allocates from the current page of the chunk.
//Scans through the VarSizeInfo objects in the page and gets the free slot
void* Chunk::allocateFromCurPageForVarSize(size_t size)
{
    //Should be called only for data items <PAGE_SIZE
    Page *page = ((PageInfo*)curPage_);
    VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)page) +sizeof(PageInfo));

    while ((unsigned char*) varInfo < ((unsigned char*)page + PAGE_SIZE))
    {
        if (0 == varInfo->isUsed_)
        {
            if( size + sizeof(VarSizeInfo) < varInfo->size_)
            {
                splitDataBucket(varInfo, size);
                return (unsigned char*)varInfo + sizeof(VarSizeInfo);
            }else if (size == varInfo->size_) {
                varInfo->isUsed_ = 1;
                return (unsigned char *) varInfo + sizeof(VarSizeInfo);
            }

        }
        varInfo = (VarSizeInfo*)((unsigned char*)varInfo
                                    + sizeof(VarSizeInfo)
                                    + varInfo->size_);
    }
    return NULL;
}

//Allocates memory to store data of variable size
void* Chunk::allocate(Database *db, size_t size, BackEnd::DbRetVal *status)
{
    if (!size) return NULL;
    //check if the size is more than PAGE_SIZE
    //if it is more than the PAGE_SIZE, then allocate new
    //page using database and then link the curPage to the
    //newly allocated page
    //if it is less than PAGE_SIZE, then check the curpage for
    //free memory of specified size
    //if not available, then scan from the firstPage for the free
    //space

    //TODO::During the scan, merge nearby nodes if both are free
    //if not available then allocate new page

    size_t alignedSize = OS::align(size);
    void *data = NULL;

    if (alignedSize > PAGE_SIZE)
    {
        /// the requested size is larger then PAGE_SIZE.
        data =  allocateForLargeDataSize(db, alignedSize);
    }
    else
    {
        /// First of all, we try to allocate from the current page.
        data = allocateFromCurPageForVarSize(alignedSize);
        if (!data) {
            //No available spaces in the current page.
            //allocate new page
            data= allocFromNewPageForVarSize(db, alignedSize);
            if (!data) {
                if (status != NULL) *status = (DbRetVal)HA_ERR_OUT_OF_MEM;
            }
        }
    }
    return data;
}

//Assumes chunk mutex is already taken, before calling this
void* Chunk::varSizeFirstFitAllocate(size_t size)
{
    Page *page = ((PageInfo*)firstPage_);
    size_t alignedSize = OS::align(size);
    while(NULL != page)
    {
        VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)page) + sizeof(PageInfo));
        while ((char*) varInfo < ((char*)page + PAGE_SIZE))
        {
            if (0 == varInfo->isUsed_)
            {
                if( alignedSize +sizeof(VarSizeInfo) < varInfo->size_)
                {
                    splitDataBucket(varInfo, alignedSize);
                    return ((char*)varInfo) + sizeof(VarSizeInfo);
                }
                else if (alignedSize == varInfo->size_) {
                    varInfo->isUsed_ = 1;
                    return ((char *) varInfo) + sizeof(VarSizeInfo);
                }
            }
            varInfo = (VarSizeInfo*)((char*)varInfo + sizeof(VarSizeInfo)
                                    +varInfo->size_);
        } /// End of While.
        page = ((PageInfo*) page)->nextPage_;  /// Loop to the next,
    }
    return NULL;
}

void Chunk::freeForVarSizeAllocator(void *ptr, int pslot)
{
    int ret = getChunkMutex(pslot);
    if (ret != 0)
    {
        return;
    }
    VarSizeInfo *varInfo = (VarSizeInfo*)((char*)ptr- sizeof(VarSizeInfo));
    varInfo->isUsed_ = 0;
    if(varInfo->size_ > (PAGE_SIZE - (sizeof(VarSizeInfo)+sizeof(PageInfo))))
   {
       PageInfo *pageInfo = (PageInfo*)((char*)varInfo - sizeof(PageInfo));
       PageInfo *pInfo = (PageInfo*)firstPage_,*prev =(PageInfo*)firstPage_;
       bool found = false;
        while(!found)
      {
        if(NULL==pInfo) break;
        if (pInfo == pageInfo) {found = true; break; }
        prev = pInfo;
        pInfo = (PageInfo*)pInfo->nextPage_;
      }
       if (!found)
      {
        releaseChunkMutex(pslot);
        return ;
      }
       if(curPage_==(Page*) pageInfo) {curPage_ = (Page*)prev ; }
       pageInfo->isUsed_ = 0;
       pageInfo->nextPageAfterMerge_ = NULL;
       pageInfo->hasFreeSpace_ = 1;
       prev->nextPage_ = pageInfo->nextPage_;
    }
    varInfo->isUsed_ = 0;
    releaseChunkMutex(pslot);
    return;

}

void Chunk::freeForLargeAllocator(void *ptr, int pslot)
{
    //There will be max only one data element in a page.
    //PageInfo is stored just before the data.
    int ret = getChunkMutex(pslot);
    if (ret != 0)
       return;

    PageInfo *pageInfo = (PageInfo*)
    (((char*)ptr) - (sizeof(PageInfo) + sizeof(int)));
    PageInfo *pInfo = (PageInfo*)firstPage_, *prev = (PageInfo*)firstPage_;
    bool found = false;
    while(!found)
    {
        if (pInfo == pageInfo) {found = true;  break; }
        prev = pInfo;
        pInfo = (PageInfo*)pInfo->nextPage_;
    }
    if (!found)
    {
        releaseChunkMutex(pslot);
        return ;
    }
    //prev->nextPage_ = pageInfo->nextPage_;
     OS::memset(((char*)pageInfo+sizeof(PageInfo)),  0 , allocSize_);
     if(((PageInfo*)firstPage_)->nextPage_ != NULL){

      pageInfo->nextPageAfterMerge_ = NULL;
      pageInfo->isUsed_ = 0;
      pageInfo->hasFreeSpace_ = 1;
      pageInfo->isMerged_   = false;
      if((Page*)pageInfo == firstPage_ && ((PageInfo*)firstPage_)->nextPage_ != NULL)
      firstPage_ = pageInfo->nextPage_ ;
      else
       prev->nextPage_ = pageInfo->nextPage_;
    }
    releaseChunkMutex(pslot);
    return;
}

//Frees the memory pointed by ptr
void Chunk::free(Database *db, void *ptr)
{
    if (!allocSize_)
    {
        freeForVarSizeAllocator(ptr,current_session->getProcSlot());
        return;
    }
    int noOfDataNodes =OS::floor((PAGE_SIZE - sizeof(PageInfo)) / allocSize_);

    if (!noOfDataNodes)
    {
      freeForLargeAllocator(ptr, current_session->getProcSlot());
      return;
    }
    int ret = getChunkMutex(current_session->getProcSlot());
    if (ret != 0)return;

     //below is the code for freeing in fixed size allocator

     //unset the used flag
    *((int*)ptr -1 ) = 0;
    PageInfo *pageInfo= getPageInfo(db, ptr);

    if (! pageInfo)
      goto RELEASE_LOCK;

    //set the pageinfo where this ptr points
    pageInfo->hasFreeSpace_ = 1;

RELEASE_LOCK:
    releaseChunkMutex(current_session->getProcSlot());
    return;
}

//returns the pageInfo of the page where this ptr points
//This works only if the data size is less than PAGE_SIZE
//If ptr points to data which is more than PAGE_SIZE,then
//calling this might lead to memory corruption
//Note:IMPORTANT::assumes db lock is taken before calling this
PageInfo* Chunk::getPageInfo(Database*, void *ptr)
{
    if (allocSize_ < PAGE_SIZE - sizeof(PageInfo)) {

       size_t rem = (size_t) ptr % COLLAPSAR_PAGE_SIZE;
       return (PageInfo*)(((char*)ptr) - rem);
    } else {
        //large size allocator
        char *inPtr = (char*)ptr;
        PageInfo* pageInfo = ((PageInfo*)firstPage_);

        while( pageInfo != NULL )
        {
          if (inPtr > (char*) pageInfo && pageInfo->nextPageAfterMerge_ >inPtr)
            return pageInfo;
          pageInfo = (PageInfo*)pageInfo->nextPage_ ;
        }
    }
    return NULL;
}

//If called on chunk used to store tuples, it returns the total number of rows
//present in the table


//TODO::for other type of allocators
int Chunk::compact()
{
    PageInfo* pageInfo = ((PageInfo*)firstPage_);
    PageInfo* prevPage = pageInfo;
    if (NULL == pageInfo)
                        {return 0;}
    pageInfo = (PageInfo*)pageInfo->nextPage_;
      if (0 == allocSize_)
    {
      while( pageInfo != NULL )
      {
        bool flag = false;
        VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)pageInfo) +
                                               sizeof(PageInfo));
        while ((char*) varInfo < ((char*)pageInfo + PAGE_SIZE))
        {
            if (1 == varInfo->isUsed_) {flag=true; break;}
            varInfo = (VarSizeInfo*)((char*)varInfo + sizeof(VarSizeInfo)
                                   +varInfo->size_);
        }
        if (!flag) {
            prevPage->nextPage_ = pageInfo->nextPage_;
            pageInfo->isUsed_ = 0;
        }
        prevPage = pageInfo;
        pageInfo = (PageInfo*)(((PageInfo*)pageInfo)->nextPage_) ;
      }
    }else if (allocSize_ < PAGE_SIZE)
    {
      while( pageInfo != NULL )
      {
        bool flag = false;
        int noOfDataNodes=(int)OS::floor((PAGE_SIZE - sizeof(PageInfo))/allocSize_);
        unsigned char *data = ((unsigned char*)pageInfo) + sizeof(PageInfo);
        for (int i = 0; i< noOfDataNodes -1; i++)
        {
            if (1 == *((int*)data)) { flag = true; break; }
            data = data +allocSize_;
        }
        if (!flag) {
            prevPage->nextPage_ = pageInfo->nextPage_;
            pageInfo->isUsed_ = 0;
        }
        prevPage = pageInfo;
        pageInfo = (PageInfo*)(((PageInfo*)pageInfo)->nextPage_) ;
      }  /// End of While
    }
    return 0;
}

size_t Chunk::totalPages()
{
    //logic is same for variable size and for large data node allocator.
    PageInfo* pageInfo = ((PageInfo*)firstPage_);
    size_t totPages=0;
    while( pageInfo != NULL )
    {
      totPages++;
      pageInfo = (PageInfo*)(((PageInfo*)pageInfo)->nextPage_) ;
    }
    return totPages;
}


void Chunk::splitDataBucket(VarSizeInfo *varInfo,
                            size_t      needSize)
{
    int remSpace = varInfo->size_ - sizeof(VarSizeInfo) - needSize;
    varInfo->isUsed_ = 1;
    varInfo->size_ = needSize;
    varInfo = (VarSizeInfo*)((char*)varInfo+ sizeof(VarSizeInfo)
                                               +  varInfo->size_);
    varInfo->isUsed_ = 0;
    varInfo->size_   = remSpace;
    return;
}


void Chunk::createDataBucket (Page *page,
                              size_t,
                              size_t needSize)
{
    VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)page) + sizeof(PageInfo));
    varInfo->isUsed_ = 0;
    varInfo->size_ = PAGE_SIZE - sizeof(PageInfo) - sizeof(VarSizeInfo);
    splitDataBucket(varInfo, needSize);
    return;
}
void Chunk::print()
{
  errmsg_printf(ERRMSG_LVL_DBUG, _(" <Chunk Id> %ld </Chunk Id> \n"),
                                                            chunkID_);
  errmsg_printf(ERRMSG_LVL_DBUG, _(" <TotalPages> %ld </TotalPages> \n"),
                                                          totalPages());
  errmsg_printf(ERRMSG_LVL_DBUG, _(" <ChunkName > %s </ChunkName> \n"),
                                                         getChunkName());
  errmsg_printf(ERRMSG_LVL_DBUG, _(" <TotalDataNodes> %ld </TotalDataNodes> \n"),
                                                     getTotalDataNodes());
  errmsg_printf(ERRMSG_LVL_DBUG, _(" <Alloc Size> %ld </Alloc Size> \n"),
                                                      getSize());
  errmsg_printf(ERRMSG_LVL_DBUG, _(" <Allocation Type> "));


   if(allocType_==0)
  {
    errmsg_printf(ERRMSG_LVL_DBUG, _("FixedSizeAllocator "));
  }else if(allocType_==1)
  {
    errmsg_printf(ERRMSG_LVL_DBUG, _("VariableSizeAllocator "));
  }else
  {
    errmsg_printf(ERRMSG_LVL_DBUG, _("UnknownAllocator "));
  }
  errmsg_printf(ERRMSG_LVL_DBUG, _("</Allocation Type>\n"));
}



